package com.chucang.shucang.ttp.consumer;

import com.chucang.shucang.common.mq.constant.KafkaConstant;
import com.chucang.shucang.common.mq.message.SmsCodeMessage;
import com.chucang.shucang.ttp.messagecenter.enums.MESSAGE_TYPE;
import com.chucang.shucang.ttp.messagecenter.enums.STRATEGY_TYPE;
import com.chucang.shucang.ttp.messagecenter.facade.MessageFacade;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author flitsneak
 * @email flitsneak@gmail.com
 * @date 2022/9/25 17:56
 * @description
 */
@Slf4j
@Component
public class MqConsumer {

    @KafkaListener(topics = {KafkaConstant.SMS_CODE_TOPIC}, errorHandler = "consumerAwareErrorHandler")
    public void smsCodeConsumer(List<ConsumerRecord<String, Object>> records, Acknowledgment ack) {

        records.forEach(s -> {
                    SmsCodeMessage smsCodeMessage = (SmsCodeMessage) s.value();
                    if (!MessageFacade.sendMessage(MESSAGE_TYPE.NORMAL, STRATEGY_TYPE.TENCENT_SMS, smsCodeMessage)) {
                        log.error("发送消息失败，消息详情为{}", smsCodeMessage);
                    }
                }
        );
        ack.acknowledge();
    }

}
